package com.garden.rocketmq.learn.demo.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 事务生产者 发送事务消息
 * 用于注重消息事务的场景
 * Created by Garden on 2020/8/24.
 */
public class TransactionProducer {

    public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, InterruptedException {
        //实例化消息生产者Producer
        TransactionMQProducer producer = new TransactionMQProducer("transaction_producer_group");
        //设置NameServer的地址，Producer启动后通过NameServer获取目标Topic的路由信息
        producer.setNamesrvAddr("localhost:9876");
        //设置线程池，
        //用于处理执行本地事务[服务端broker存储haft消息成功后请求]或者检查本地事务逻辑[服务端broker接收Pending消息后回查请求]提供线程
        ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("client-transaction-msg-check-thread");
                return thread;
            }
        });
        producer.setExecutorService(executorService);
        //设置本地事务监听器，用于处理执行本地事务或者检查本地事务逻辑
        producer.setTransactionListener(new TransactionListener(){
            private AtomicInteger transactionIndex = new AtomicInteger(0);
            private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //执行本地事务
                //记录消息事务id以及自定义状态值，返回Pending状态触发回查机制
                int status = (transactionIndex.getAndIncrement()) % 3;
                localTrans.put(message.getTransactionId(),status);
                return LocalTransactionState.UNKNOW;
            }
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                //检查本地事务
                //从本地事务集合中获取目标消息事务对应的自定义状态值，根据自定义状态值返回事务状态
                Integer status = localTrans.get(messageExt.getTransactionId());
                if (null != status) {
                    switch (status) {
                        case 0:
                            return LocalTransactionState.UNKNOW;
                        case 1:
                            return LocalTransactionState.COMMIT_MESSAGE;
                        case 2:
                            return LocalTransactionState.ROLLBACK_MESSAGE;
                        default:
                    }
                }
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        //启动Producer实例
        producer.start();

        //发送消息
        for (int i = 0; i < 10; i++){
            //创建消息，指定主题Topic，标签Tag，业务标识Key，消息体MsgBody [Tag可用于消息过滤，Key可用于消息查询]
            Message msg = new Message("test_topic","transaction_producer_tag"
                    ,"transaction_producer_key." + i, ("TransactionProducer Msg." + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            //发送消息至Broker
            SendResult sendResult = producer.sendMessageInTransaction(msg,null);
            //通过发送结果查看是否发送成功
            System.out.printf("%s%n", sendResult);
        }

        //睡眠主线程，事务生产者处理执行本地事务以及检查本地事务
        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }

        //发送完毕，关闭Producer实例
        producer.shutdown();
    }

}
